package com.atguigu.day06;

import com.atguigu.bean.Event;
import com.atguigu.day03.Flink01_Source_Customer;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;

public class Flink07_ProcessAllWindowTopN {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.获取数据
        SingleOutputStreamOperator<Event> streamSource = env.addSource(new Flink01_Source_Customer.ClickSource())
                //指定WaterMark并分配时间戳
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override
                            public long extractTimestamp(Event element, long recordTimestamp) {
                                return element.timestamp;
                            }
                        })
                );

        //3.将之前流里面的url提取出来
        SingleOutputStreamOperator<String> urlStream = streamSource.map(new MapFunction<Event, String>() {
            @Override
            public String map(Event value) throws Exception {
                return value.url;
            }
        });

        //5.需要统计最近10秒钟内最热门的两个url链接，并且每5秒钟更新一次
        urlStream.windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                .process(new MyTopNURL(3))
                .print();

        env.execute();
    }

    public static class MyTopNURL extends ProcessAllWindowFunction<String, Tuple2<String,Integer>, TimeWindow>{

        private Integer topn;

        public MyTopNURL(Integer topN){
            this.topn = topN;
        }


        @Override
        public void process(Context context, Iterable<String> elements, Collector<Tuple2<String, Integer>> out) throws Exception {
            //1.创建hashMap用来统计不同的url的个数
            HashMap<String, Integer> hashMap = new HashMap<>();
            //2.遍历出迭代器中的每个数据（最近10S的url）
            for (String url : elements) {
                if (hashMap.containsKey(url)){
                    //取出之前统计的个数
                    Integer lastCount = hashMap.get(url);
                    lastCount = lastCount + 1;
                    hashMap.put(url, lastCount);
                }else {
                    //url第一次来
                    hashMap.put(url, 1);
                }

            }

            //3.遍历出hashMap中的数据进行排序
            //创建List集合用来排序
            ArrayList<Tuple2<String, Integer>> urlToCountList = new ArrayList<>();
            for (String url : hashMap.keySet()) {
                Integer value = hashMap.get(url);
                urlToCountList.add(Tuple2.of(url, value));
            }

            //4.对集合中的数据从大到小排序
            urlToCountList.sort(new Comparator<Tuple2<String, Integer>>() {
                @Override
                public int compare(Tuple2<String, Integer> o1, Tuple2<String, Integer> o2) {
                    return o2.f1-o1.f1;
                }
            });

            //5.取出点击最高两个url
            for (int i = 0; i < topn; i++) {
                out.collect(urlToCountList.get(i));
            }

        }
    }
}
